/*****************************************************************************
 * Copyright [2017] [taurus.ai]
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *  http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 *****************************************************************************/

/**
 * ITDEngine: base class of all trade engine.
 * @Author cjiang (changhao.jiang@taurus.ai)
 * @since   April, 2017
 */

#include "ITDEngine.h"
#include "PageCommStruct.h" /**< REQUEST_ID_RANGE */
#include "Timer.h"
#include "longfist/LFConstants.h"
#include "longfist/LFUtils.h"
#include "longfist/sys_messages.h"
#include <algorithm>
#include <boost/bind.hpp>

USING_WC_NAMESPACE

ITDEngine::ITDEngine(short source): IEngine(source), request_id(1), local_id(1), rid(1)
{}

void ITDEngine::set_reader_thread()
{
    reader_thread = ThreadPtr(new std::thread(boost::bind(&ITDEngine::listening, this)));
}


void MsgReadHandler::OnRtnStatus(struct RtnStatusItem *item)
{
    LFRtnOrderField md_order = {};
    engine->on_rtn_order(&md_order);
}

void MsgReadHandler::OnRtnTrade(struct RtnTradeItem *item)
{
    LFRtnTradeField md_trade = {};
    engine->on_rtn_trade(&md_trade);
}

vector<string> ss_split(const string &str, char delimiter )
{
    std::stringstream ss(str);
    string temp;
    vector<string> rlt;
    while(std::getline(ss, temp, delimiter))
    {
        rlt.push_back(temp);
    }
    return rlt;

}


bool ITDEngine::setup_socket_udp(int *sock, const char *local_ip, const char *mcast_ip, uint16_t mcast_port)
{
    int ret = 0;
    *sock = ret = socket(AF_INET, SOCK_DGRAM, 0);
    if (ret == -1) {
        KF_LOG_ERROR(engine->logger, "[setup_udp] socket error");
        return false;
    }

    const int on = 1;
    ret = setsockopt(*sock, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));
    if (ret == -1) {
        KF_LOG_ERROR(engine->logger, "[setup_udp] set socket opt error: reuse");
        return false;
    }
    struct sockaddr_in local_addr;
    memset(&local_addr, 0, sizeof(local_addr));
    local_addr.sin_family = AF_INET;
    inet_aton(mcast_ip, &local_addr.sin_addr);
    local_addr.sin_port = htons(mcast_port);
    ret = bind(*sock, (struct sockaddr *) &local_addr, sizeof(local_addr));
    if (ret == -1) {
        KF_LOG_ERROR(engine->logger, "[setup_udp] bind error");
        return false;
    }

    struct ip_mreq mreq;
    mreq.imr_interface.s_addr = inet_addr(local_ip);
    mreq.imr_multiaddr.s_addr = inet_addr(mcast_ip);
    ret = setsockopt(*sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq, sizeof(mreq));
    if (ret == -1) {
        KF_LOG_ERROR(engine->logger, "[setup_udp] set socket opt error: mcast");
        return false;
    }
    return true;
}



void ITDEngine::RtnMsgReaderRem(ITDEngine  *engine, int cpu_num) 
{
        
    struct epoll_event ev, events[MAX_EVENTS];
    int epfd = -1, nfds = 0;
    unordered_map<int,int> udp_sockets;

    char buf[MSG_MAX_LEN];
    int64_t ret;

    ev.events = EPOLLIN;
    epfd = epoll_create(engine->mcast_uris.size());
    if (epfd == -1) {
        KF_LOG_ERROR(engine->logger, "[main] epoll create failed");
        return -1;
    }
    if (accounts.size() != engine->mcast_uris.size() ) 
    {
        KF_LOG_ERROR(engine->logger, "accounts and udp--- not match");
        return;
    }
    for (size_t i = 0; i < engine->mcast_uris.size(); ++i) {
        auto& m = engine->mcast_uris[i];

        int sock_udp = -1;
        if (!setup_socket_udp(&sock_udp, engine->local_address.c_str(), m.first.c_str(), m.second)) {
            KF_LOG_ERROR(engine->logger, "[main] setup udp socket failed: i=" << i << " mcast_address=" << m.first << " mcast_port=" << m.second);
        }
        ev.data.fd = sock_udp;
        if (epoll_ctl(epfd, EPOLL_CTL_ADD, sock_udp, &ev) < 0) {
            KF_LOG_ERROR(engine->logger, "[main] epoll ctl add failed: i=" << i << " mcast_address=" << m.first << " mcast_port=" << m.second);
            return -1;
        }
        udp_sockets[sock_udp]=i;
        KF_LOG_INFO(engine->logger, "[main] listening to udp socket: i=" << i << " mcast_address=" << m.first << " mcast_port=" << m.second);
    }

    KF_LOG_INFO(engine->logger, "[main] start receiving data");
    while (true) {
        nfds = epoll_wait(epfd, events, MAX_EVENTS, 0);
        for (int i = 0; i != nfds; ++i) {
            int recv_fd = events[i].data.fd;
            ret = recv(recv_fd, buf, MSG_MAX_LEN, 0);
            if (ret == 0) {
                KF_LOG_ERROR(engine->logger, "[process_udp] connection closed");
                break;
            } 
            else if (ret < 0) {
                KF_LOG_ERROR(engine->logger, "[process_udp] recv market data error: ret=" << ret);
                break;
            }
            int len = 0;
            int  account_index = udp_sockets[recv_fd];
            while (ret - len > 0) {
                len += rtn_call(buf + len,  account_index);
            }
        }
    }

    for (auto ele : udp_sockets) {
        close(ele.first);
    }

}



int  ITDEngine::rtn_call(char *data, int account_index )
{
    struct  order_udp_rem *udp_rem=(order_udp_rem*) data;
   
    int len=sizeof(order_udp_rem);
    // #define EES_OrderStatus_executed                0x20
    if (udp_rem->status == 0x20)
    {
    // trade 
        LFRtnTradeField md_trade = {};
       // md_trade.order_id = udp_rem->
        this->on_rtn_trade(&md_trade);
    }
    else
    {
        // order
        LFRtnOrderField md_order = {};
        this->on_rtn_order(&md_order);
    }
    return len;
}


void ITDEngine::RtnMsgReaderYD(ITDEngine  *engine, int cpu_num) 
{
    if ( (engine->reader_users.size() != engine->reader_pwds.size())  || engine->reader_users.size()==0  )
        return ;

    vector<string> user( ss_split(engine->reader_users, ',') );
    vector<string> pwd( ss_split(engine->reader_pwds, ',') );
    vector<string> udp_front( ss_split(engine->reader_udp_fronts, ','));
    
    vector<RtnMsgReaderApi*> users_api;
    
    for (unsigned int i=0; i< engine->reader_users.size(); i++)
    {
        users_api.push_back(RtnMsgReaderApi::CreateRtnMsgReaderApi());
    }
    
    int i=0;
    auto itr = users_api.begin();
    for (; itr!= users_api.end(); itr++)
    {
       i++;
       int ret= (*itr)->Login(user[i].c_str(), pwd[i].c_str(), udp_front[i].c_str());
       if (!ret)
       {
         KF_LOG_DEBUG(engine->logger, "RtnMsgReader" << "user:" << user[i]<<"login failed");
         itr=users_api.erase(itr);
       }
    }
    engine->cpupin(cpu_num);
    MsgReadHandler handle(engine);
    while (true)
    {
         if ( !(engine->get_running()))
         {
             break;
         }
         for (unsigned int j=0; j<users_api.size(); j++)
         {
            users_api[j]->PollMsg(&handle);
         }
    }

}




void ITDEngine::init()
{
    reader = yijinjing::JournalReader::createRevisableReader(name());
    JournalPair tdPair = getTdJournalPair(source_id);
    writer = yijinjing::JournalSafeWriter::create(tdPair.first, tdPair.second, name());
    JournalPair tdSendPair = getTdSendJournalPair(source_id);
    send_writer = yijinjing::JournalWriter::create(tdSendPair.first, tdSendPair.second, "SEND_" + name());
    client_infos = TDClientInfoHelperPtr(new TDClientInfoHelper(source_id));
    msg_queue=new rigtorp::SPSCQueue<LFInputOrderField>(9999);
    std::thread md_select_thread(&ITDEngine::RtnMsgReader,this,this,1);
    md_select_thread.detach();
}

void ITDEngine::listening()
{
    yijinjing::FramePtr frame;
    while (isRunning && signal_received < 0)
    {
        frame = reader->getNextFrame();
        if (frame.get() != nullptr)
        {
            short msg_type = frame->getMsgType();
            short msg_source = frame->getSource();
            cur_time = frame->getNano();
            if (msg_type == MSG_TYPE_LF_MD)
            {
                void* fdata = frame->getData();
                LFMarketDataField* md = (LFMarketDataField*)fdata;
                on_market_data(md, cur_time);
                continue;
            }
            else if (msg_type < 200)
            {
                // system related...
                if (msg_type == MSG_TYPE_TRADE_ENGINE_LOGIN && msg_source == source_id)
                {
                    try
                    {
                        string content((char*)frame->getData());
                        json j_request = json::parse(content);
                        string client_name = j_request["name"].get<string>();
                        if (add_client(client_name, j_request))
                            KF_LOG_INFO(logger, "[user] accepted: " << client_name);
                        else
                            KF_LOG_INFO(logger, "[user] rejected: " << client_name);
                    }
                    catch (...)
                    {
                        KF_LOG_ERROR(logger, "error in parsing TRADE_ENGINE_LOGIN: " << (char*)frame->getData());
                   }
                }
                else if (msg_type == MSG_TYPE_STRATEGY_END)
                {
                    try
                    {
                        string content((char*)frame->getData());
                        json j_request = json::parse(content);
                        string client_name = j_request["name"].get<string>();
                        if (remove_client(client_name, j_request))
                            KF_LOG_INFO(logger, "[user] removed: " << client_name);
                    }
                    catch (...)
                    {
                        KF_LOG_ERROR(logger, "error in parsing STRATEGY_END: " << (char*)frame->getData());
                    }
                }
                else if (msg_type == MSG_TYPE_STRATEGY_POS_SET && msg_source == source_id)
                {
                    try
                    {
                        string content((char*)frame->getData());
                        json j_request = json::parse(content);
                        string client_name = j_request["name"].get<string>();
                        client_infos->set_pos(client_name, j_request);
                        KF_LOG_INFO(logger, "[user] set pos: client=" << client_name << " pos=" << content);
                    }
                    catch (...)
                    {
                        KF_LOG_ERROR(logger, "error in parsing STRATEGY_POS_SET: " << (char*)frame->getData());
                    }
                }
                else if (msg_type == MSG_TYPE_STRATEGY_ALPHA_SET && msg_source == source_id)
                {
                    try
                    {
                        string content((char*)frame->getData());
                        json j_request = json::parse(content);
                        string client_name = j_request["name"].get<string>();
                        client_infos->set_alpha(client_name, j_request, false);
                        j_request.erase("name");
//                      KF_LOG_INFO(logger, "[user] update alpha: client=" << client_name << " alpha=" << j_request.dump());
                    }
                    catch (...)
                    {
                        KF_LOG_ERROR(logger, "error in parsing STRATEGY_ALPHA_SET: " << (char*)frame->getData());
                    }
                }
                else if (msg_type == MSG_TYPE_TRADE_ENGINE_OPEN && (msg_source <= 0 || msg_source == source_id))
                {
                    on_engine_open();
                }
                else if (msg_type == MSG_TYPE_TRADE_ENGINE_CLOSE && (msg_source <= 0 || msg_source == source_id))
                {
                    on_engine_close();
                }
                else if (msg_type == MSG_TYPE_STRING_COMMAND)
                {
                    string cmd((char*)frame->getData());
                    on_command(cmd);
                }
                else if (msg_type == MSG_TYPE_SWITCH_TRADING_DAY)
                {
                    client_infos->switch_day();
                    local_id = 1;
                    on_switch_day();
                }
            }
            else if (msg_source == source_id && is_logged_in())
            {
                // from client
                string name = reader->getFrameName();
                auto iter = clients.find(name);
                if (iter == clients.end())
                    continue;

                void* fdata = frame->getData();
                int requestId = frame->getRequestId();
                switch (msg_type)
                {
                    case MSG_TYPE_LF_QRY_POS:
                    {
                        LFQryPositionField* data = (LFQryPositionField*)fdata;
                        req_investor_position(data, data->account_index, requestId);
                        break;
                    }
                    case MSG_TYPE_LF_QRY_ACCOUNT:
                    {
                        LFQryAccountField* data = (LFQryAccountField*)fdata;
                        req_qry_account(data, data->account_index, requestId);
                        break;
                    }
                    case MSG_TYPE_LF_QRY_SETTLEMENT:
                    {
                        LFQrySettlementField* data = (LFQrySettlementField*)fdata;
                        req_qry_settlement(data, data->account_index, requestId);
                        break;
                    }
                    case MSG_TYPE_LF_ORDER:
                    {
                        LFInputOrderField* order = (LFInputOrderField*)fdata;

                        // the increment of local_id has been moved to req_order_insert()
                        record_order(local_id, order->order_id, requestId);
                        order->local_id = local_id;

                        long before_nano = kungfu::yijinjing::getNanoTime();
                        req_order_insert(order, order->account_index, requestId, cur_time);
                        // insert order, we need to track in send
                        send_writer->write_frame_extra(order, sizeof(LFInputOrderField), source_id, MSG_TYPE_LF_ORDER, true, requestId, before_nano);
                        KF_LOG_DEBUG(logger, "[insert_order]" << " account_index=" << order->account_index
                                                              << " order_id=" << order->order_id
                                                              << " local_id=" << order->local_id
                                                              << " sys_id=" << order->sys_id
                                                              << " ticker=" << order->ticker
                                                              << " rid=" << requestId
                                                              );
                        break;
                    }
                    case MSG_TYPE_LF_ORDER_ACTION:
                    {
                        LFOrderActionField* order = (LFOrderActionField*)fdata;
                        req_order_action(order, order->account_index, requestId, cur_time);
                        KF_LOG_DEBUG(logger, "[cancel_order]" << " account_index=" << order->account_index
                                                              << " order_id=" << order->order_id
                                                              << " local_id=" << order->local_id
                                                              << " sys_id=" << order->sys_id
                                                              << " rid=" << requestId
                                                              );
                        break;
                    }
                    case MSG_TYPE_LF_QRY_INSTRUMENT:
                    {
                        LFQryInstrumentField* data = (LFQryInstrumentField*)fdata;
                        req_qry_instrument(data, data->account_index, requestId);
                        break;
                    }
                    case MSG_TYPE_LF_QRY_INSTRUMENT_COMMISSION_RATE:
                    {
                        LFQryInstrumentField* data = (LFQryInstrumentField*)fdata;
                        req_qry_instrument_commission_rate(data, data->account_index, requestId);
                        break;
                    }
                    case MSG_TYPE_LF_QRY_INSTRUMENT_MARGIN_RATE:
                    {
                        LFQryInstrumentField* data = (LFQryInstrumentField*)fdata;
                        req_qry_instrument_margin_rate(data, data->account_index, requestId);
                        break;
                    }
                    default:
                        KF_LOG_DEBUG(logger, "[Unexpected] frame found: msg_type=" << msg_type << ", name=" << name);
                }
            }
        }
    }

    if (IEngine::signal_received >= 0)
    {
        KF_LOG_INFO(logger, "[IEngine] signal received: " << IEngine::signal_received);
    }

    if (!isRunning)
    {
        KF_LOG_INFO(logger, "[IEngine] forced to stop.");
    }
}

static const LFRspPositionField empty_pos = {};

void ITDEngine::on_rsp_position(const LFRspPositionField* pos, bool isLast, int requestId, int errorId, const char* errorMsg)
{
    if (errorId == 0) {
        writer->write_frame(pos, sizeof(LFRspPositionField), source_id, MSG_TYPE_LF_RSP_POS, isLast, requestId);
        KF_LOG_DEBUG(logger, "[on_rsp_position]" << " account_index=" << pos->account_index 
                                                 << " ticker=" << pos->InstrumentID
                                                 << " dir=" << pos->PosiDirection
                                                 << " pos=" << pos->Position
                                                 << " ytd=" << pos->YdPosition
                                                 << " rid=" << requestId);
    }
    else {
        writer->write_error_frame(pos, sizeof(LFRspPositionField), source_id, MSG_TYPE_LF_RSP_POS, isLast, requestId, errorId, errorMsg);
        KF_LOG_ERROR(logger, "[on_rsp_position]" << " rid=" << requestId << " error_id=" << errorId << " error_msg=" << errorMsg);
    }
}

void ITDEngine::on_rsp_account(const LFRspAccountField* account, bool isLast, int requestId, int errorId, const char* errorMsg)
{
    if (errorId == 0) {
        writer->write_frame(account, sizeof(LFRspAccountField), source_id, MSG_TYPE_LF_RSP_ACCOUNT, isLast, requestId);
        KF_LOG_DEBUG(logger, "[on_rsp_account]" << " account_index=" << account->account_index 
                                                << " rid=" << requestId);
    }
    else {
        writer->write_error_frame(account, sizeof(LFRspAccountField), source_id, MSG_TYPE_LF_RSP_ACCOUNT, isLast, requestId, errorId, errorMsg);
        KF_LOG_ERROR(logger, "[on_rsp_account]" << " rid=" << requestId << " error_id=" << errorId << " error_msg=" << errorMsg);
    }
}

void ITDEngine::on_rsp_settlement(const LFRspSettlementField* settle, bool isLast, int requestId, int errorId, const char* errorMsg)
{
    if (errorId == 0) {
        writer->write_frame(settle, sizeof(LFRspSettlementField), source_id, MSG_TYPE_LF_RSP_SETTLEMENT, isLast, requestId);
        KF_LOG_DEBUG(logger, "[on_rsp_settlement]" << " account_index=" << settle->account_index 
                                                   << " trading_day=" << settle->TradingDay 
                                                   << " investor=" << settle->InvestorID
                                                   << " rid=" << requestId);
    }
    else {
        writer->write_error_frame(settle, sizeof(LFRspSettlementField), source_id, MSG_TYPE_LF_RSP_SETTLEMENT, isLast, requestId, errorId, errorMsg);
        KF_LOG_ERROR(logger, "[on_rsp_settlement]" << " rid=" << requestId << " error_id=" << errorId << " error_msg=" << errorMsg);
    }
}

void ITDEngine::on_rsp_instrument(const LFRspInstrumentField* ins, bool isLast, int requestId, int errorId, const char* errorMsg)
{
    if (errorId == 0) {
        writer->write_frame(ins, sizeof(LFRspInstrumentField), source_id, MSG_TYPE_LF_RSP_INSTRUMENT, isLast, requestId);
//      KF_LOG_DEBUG(logger, "[on_rsp_instrument]" << " rid=" << requestId << " instrument=" << ins->ticker);
    }
    else {
        writer->write_error_frame(ins, sizeof(LFRspInstrumentField), source_id, MSG_TYPE_LF_RSP_INSTRUMENT, isLast, requestId, errorId, errorMsg);
        KF_LOG_ERROR(logger, "[on_rsp_instrument]" << " rid=" << requestId 
                                                   << " error_id=" << errorId 
                                                   << " error_msg=" << errorMsg
                                                   );
    }
}

void ITDEngine::on_rsp_instrument_commission_rate(const LFRspInstrumentCommissionRateField* ins, bool isLast, int requestId, int errorId, const char* errorMsg)
{
    if (errorId == 0) {
        writer->write_frame(ins, sizeof(LFRspInstrumentCommissionRateField), source_id, MSG_TYPE_LF_RSP_INSTRUMENT_COMMISSION_RATE, isLast, requestId);
//      KF_LOG_DEBUG(logger, "[on_rsp_instrument_commission]" << " rid=" << requestId << " instrument=" << ins->ticker);
    }
    else {
        writer->write_error_frame(ins, sizeof(LFRspInstrumentCommissionRateField), source_id, MSG_TYPE_LF_RSP_INSTRUMENT_COMMISSION_RATE, isLast, requestId, errorId, errorMsg);
        KF_LOG_ERROR(logger, "[on_rsp_instrument_commission]" << " rid=" << requestId 
                                                              << " error_id=" << errorId 
                                                              << " error_msg=" << errorMsg
                                                              );
    }
}

void ITDEngine::on_rsp_instrument_margin_rate(const LFRspInstrumentMarginRateField* ins, bool isLast, int requestId, int errorId, const char* errorMsg)
{
    if (errorId == 0) {
        writer->write_frame(ins, sizeof(LFRspInstrumentMarginRateField), source_id, MSG_TYPE_LF_RSP_INSTRUMENT_MARGIN_RATE, isLast, requestId);
//      KF_LOG_DEBUG(logger, "[on_rsp_instrument_margin]" << " rid=" << requestId << " ticker=" << ins->ticker);
    }
    else {
        writer->write_error_frame(ins, sizeof(LFRspInstrumentMarginRateField), source_id, MSG_TYPE_LF_RSP_INSTRUMENT_MARGIN_RATE, isLast, requestId, errorId, errorMsg);
        KF_LOG_ERROR(logger, "[on_rsp_instrument_margin]" << " rid=" << requestId 
                                                          << " error_id=" << errorId 
                                                          << " error_msg=" << errorMsg
                                                          );
    }
}

void ITDEngine::on_rsp_order_insert(LFInputOrderField* order, int errorId, const char* errorMsg)
{
    tie(order->order_id, rid) = get_order_request_id(order->local_id);
    if (errorId == 0) {
        writer->write_frame(order, sizeof(LFInputOrderField), source_id, MSG_TYPE_LF_ORDER, true, rid);
        KF_LOG_DEBUG(logger, "[on_rsp_order]" << " order_id=" << order->order_id 
                                              << " local_id=" << order->local_id 
                                              << " sys_id=" << order->sys_id
                                              << " rid=" << rid
                                              );
    }
    else {
        writer->write_error_frame(order, sizeof(LFInputOrderField), source_id, MSG_TYPE_LF_ORDER, true, rid, errorId, errorMsg);
        KF_LOG_ERROR(logger, "[on_rsp_order]" << " order_id=" << order->order_id 
                                              << " local_id=" << order->local_id 
                                              << " sys_id=" << order->sys_id
                                              << " error_id=" << errorId 
                                              << " error_msg=" << errorMsg
                                              << " rid=" << rid
                                              );
    }
}

void ITDEngine::on_rsp_order_action(LFOrderActionField *action, int errorId, const char *errorMsg)
{
    tie(action->order_id, rid) = get_order_request_id(action->local_id);
    if (errorId == 0) {
        writer->write_frame(action, sizeof(LFOrderActionField), source_id, MSG_TYPE_LF_ORDER_ACTION, true, rid);
        KF_LOG_DEBUG(logger, "[on_rsp_action]" << " order_id=" << action->order_id 
                                               << " local_id=" << action->local_id 
                                               << " sys_id=" << action->sys_id
                                               << " rid=" << rid
                                               );
    }
    else {
        writer->write_error_frame(action, sizeof(LFOrderActionField), source_id, MSG_TYPE_LF_ORDER_ACTION, true, rid, errorId, errorMsg);
        KF_LOG_ERROR(logger, "[on_rsp_action]" << " order_id=" << action->order_id 
                                               << " local_id=" << action->local_id 
                                               << " sys_id=" << action->sys_id 
                                               << " error_id=" << errorId 
                                               << " error_msg=" << errorMsg
                                               << " rid=" << rid
                                               );
    }
}

void ITDEngine::on_rtn_order(LFRtnOrderField* rtn_order)
{
   
    order_mutex->lock();
    if(rtn_order_id.count(rtn_order->sys_id) != 0)
        return;
    rtn_order_id.insert(rtn_order->sys_id);
    order_mutex->unlock();

    tie(rtn_order->order_id, rid) = get_order_request_id(rtn_order->local_id);
    writer->write_frame(rtn_order, sizeof(LFRtnOrderField), source_id, MSG_TYPE_LF_RTN_ORDER, true, rid);
    KF_LOG_DEBUG(logger, "[on_rtn_order]" << " order_id=" << rtn_order->order_id
                                          << " local_id=" << rtn_order->local_id
                                          << " sys_id=" << rtn_order->sys_id
                                          << " status=" << rtn_order->order_status
                                          << " decre=" << rtn_order->decre
                                          << " rid=" << rid
                                          );
}

void ITDEngine::on_rtn_trade(LFRtnTradeField* rtn_trade)
{

    trade_mutex->lock();
    if(rtn_trade_id.count(rtn_trade->sys_id) != 0)
        return ;
    rtn_trade_id.insert(rtn_trade->sys_id); 
    trade_mutex->unlock();

    tie(rtn_trade->order_id, rid) = get_order_request_id(rtn_trade->local_id);
    writer->write_frame(rtn_trade, sizeof(LFRtnTradeField), source_id, MSG_TYPE_LF_RTN_TRADE, true, rid);
    KF_LOG_DEBUG(logger, "[on_rtn_trade]" << " order_id=" << rtn_trade->order_id
                                          << " local_id=" << rtn_trade->local_id
                                          << " sys_id=" << rtn_trade->sys_id
                                          << " price=" << rtn_trade->price
                                          << " volume=" << rtn_trade->volume
                                          << " rid=" << rid
                                          );

}

bool ITDEngine::add_client(const string& client_name, const json& j_request)
{
    string folder = j_request["folder"].get<string>();
    int rid_s = j_request["rid_s"].get<int>();
    int rid_e = j_request["rid_e"].get<int>();
    long last_switch_day = j_request["last_switch_nano"].get<long>();
    client_infos->load(client_name);
    auto iter = clients.find(client_name);
    if (iter == clients.end())
    {
        int idx = reader->addJournal(folder, client_name);
        reader->seekTimeJournal(idx, cur_time);
        ClientInfoUnit& status = clients[client_name];
        status.is_alive = true;
        status.journal_index = idx;
        status.rid_start = rid_s;
        status.rid_end = rid_e;
    }
    else if (iter->second.is_alive)
    {
        KF_LOG_ERROR(logger, "login already exists... (info)" << client_name);
    }
    else
    {
        if (iter->second.journal_index < 0)
            iter->second.journal_index = reader->addJournal(folder, client_name);

        reader->seekTimeJournal(iter->second.journal_index, cur_time);
        iter->second.rid_start = rid_s;
        iter->second.rid_end = rid_e;
        iter->second.is_alive = true;
    }
    /**
     * json_ack: {'name': 'bl_test', 'Pos':{'ic1701':[100,50,0,0,0,0]}, 'Source': 1, 'ok': true, 'date': 20190102}
     * failed: {'name': 'bl_test', 'ok': false}
     */
    json json_ack = client_infos->get_pos(client_name);
    if (json_ack["ok"].get<bool>())
    {
        PosHandlerPtr pos_handler = PosHandler::create(source_id, json_ack);
        if (json_ack.find("nano") != json_ack.end() && json_ack["nano"].get<long>() < last_switch_day)
        {
            pos_handler->switch_day();
            json_ack["Pos"] = pos_handler->to_json()["Pos"];
        }
    }
    json_ack["date"] = trading_day;
    json_ack["alpha"] = client_infos->get_alpha(client_name);
    for (auto& acct : accounts) {
        json_ack["accounts"].push_back(acct.UserID);
    }
    string json_content = json_ack.dump();
    writer->write_frame(json_content.c_str(), json_content.length() + 1, source_id, MSG_TYPE_TRADE_ENGINE_ACK, 1, -1);
    return true;
}

bool ITDEngine::remove_client(const string &client_name, const json &j_request)
{
    auto iter = clients.find(client_name);
    if (iter == clients.end())
        return false;
    reader->expireJournal(iter->second.journal_index);
    iter->second.is_alive = false;
    /*
    if (is_logged_in())
    {
        // cancel all pending orders, and clear the memory
        auto orders = client_infos->get_existing_orders(client_name);
        int idx = iter->second.account_index;
        for (int order_id: orders)
        {
            LFOrderActionField action = {};
            action.ActionFlag = LF_CHAR_Delete;
            action.KfOrderID = order_id;
            action.LimitPrice = 0;
            action.VolumeChange = 0;
            strcpy(action.BrokerID, accounts[idx].BrokerID);
            strcpy(action.InvestorID, accounts[idx].InvestorID);
            int local_id;
            if (client_infos->get_order(client_name, order_id, local_id, action.InstrumentID))
            {
                string order_ref = std::to_string(local_id);
                strcpy(action.OrderRef, order_ref.c_str());
                KF_LOG_DEBUG(logger, "[cancel_remain_order] rid=" << order_id << " ticker=" << action.InstrumentID << " ref=" << order_ref);
                req_order_action(&action, iter->second.account_index, order_id, cur_time);
            }
        }
    }
    */
    client_infos->remove(client_name);
    return true;
}

pair<int, int> ITDEngine::get_order_request_id(int local_id)
{
//  IDMap::iterator it = id_map.find(local_id);
//  return it == id_map.end() ? std::make_pair(-1, 0) : it->second;
    return id_arr[local_id % MAX_ORDER_LIMIT];
}

void ITDEngine::record_order(int local_id, int order_id, int request_id)
{
//  id_map[local_id] = make_pair(order_id, request_id);
    id_arr[local_id % MAX_ORDER_LIMIT] = make_pair(order_id, request_id);
}

void ITDEngine::load(const json& j_config)
{
    auto iter = j_config.find("accounts");
    if (iter != j_config.end())
    {
        int account_num = iter.value().size();
        KF_LOG_INFO(logger, "[load_account] number: " << account_num);
        // ITDEngine's resize
        accounts.resize(account_num);
        // base class resize account info structures.
        resize_accounts(account_num);
        int account_idx = 0;
        for (auto& j_account: iter.value())
        {
            accounts[account_idx] = load_account(account_idx, j_account);
            /*
            const json& j_clients = j_account["clients"];
            for (auto& j_client: j_clients)
            {
                string client_name = j_client.get<string>();
                ClientInfoUnit& client_status = clients[client_name];
                client_status.account_index = account_idx;
                client_status.is_alive = false;
                client_status.journal_index = -1;
                KF_LOG_INFO(logger, "[client] name=" << client_name << " account=" << accounts[client_status.account_index].InvestorID);
            }
            */
            // add
            account_idx ++;
        }
        
        /*load udp rem addr and port  */
        json j_mcast = j_config["McastUris"];
        for (size_t i = 0; i < j_mcast.size(); ++i) {
            vector<string> st;
            boost::algorithm::split(st, j_mcast[i].get<string>(), boost::algorithm::is_any_of(":"));
            mcast_uris.emplace_back(make_pair(st[0], boost::lexical_cast<int>(st[1])));
        }
        KF_LOG_INFO(logger,  "rem-num_mcasts=" << mcast_uris.size());
    };


    }
}



TradeAccount ITDEngine::load_account(int idx, const json& j_account)
{
    KF_LOG_ERROR(logger, "[account] NOT IMPLEMENTED! content=" << j_account);
    throw std::runtime_error("load_account not implemented yet!");
}
